package com.yk.spark.service.manage.submit.impl;

import com.yk.spark.model.AppModel;
import com.yk.spark.model.app.state.action.AppActionState;
import com.yk.spark.model.submit.SubmitOptions;
import com.yk.spark.service.file.api.IFileService;
import com.yk.spark.service.manage.submit.api.IAppSubmitService;
import org.apache.spark.deploy.rest.CreateSubmissionRequest;
import org.apache.spark.deploy.rest.RestSubmissionClient;
import org.apache.spark.deploy.rest.SubmitRestProtocolResponse;
import org.json.JSONArray;
import org.springframework.stereotype.Service;
import scala.collection.JavaConverters;
import scala.collection.immutable.Map$;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * Created by Limit on 2017/8/23.
 */
@Service("appSubmitService")
public class AppSubmitServiceImpl implements IAppSubmitService {

    private static String SPARK_MASTER="spark://Master.Hadoop:6066";
    @Resource(name = "fileService")
    private IFileService fileService;
    /**
     * 提交spark app
     *
     * @param model  app实例
     * @return app提交状态
     */
    @Override
    public AppActionState submitSparkApp(AppModel model) {

        RestSubmissionClient client=new RestSubmissionClient(SPARK_MASTER);
        CreateSubmissionRequest request=new CreateSubmissionRequest();
        SubmitOptions submitOptions=new SubmitOptions(model,this.fileService);
        request.appArgs_$eq(submitOptions.getAppArgs());
        request.environmentVariables_$eq(this.transform(submitOptions.getEnvironmentVariables()));
        request.sparkProperties_$eq(this.transform(submitOptions.getSparkProperties()));
        request.mainClass_$eq(submitOptions.getMainClass());
        request.appResource_$eq(submitOptions.getAppResource());
        request.clientSparkVersion_$eq("2.2.0");
        request.doValidate();
        SubmitRestProtocolResponse response=client.createSubmission(request);
        response.doValidate();
        return new AppActionState(response.toJson());
    }

    private scala.collection.immutable.Map<String, String> transform(Map<String,String> map){
        scala.collection.mutable.Map<String, String> mutableMap = JavaConverters.mapAsScalaMapConverter(map).asScala();
        Object obj = Map$.MODULE$.<String,String>newBuilder().$plus$plus$eq(mutableMap.toSeq());
        Object result = ((scala.collection.mutable.Builder) obj).result();
        return (scala.collection.immutable.Map)result;
    }
}
